package com.idgenerator.generator;

import com.idgenerator.utils.IpUtils;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

@Setter
@Getter
@NoArgsConstructor
public class ZookeeperIdGenerator implements IdGenerator {

    private static final Logger log = LoggerFactory.getLogger(ZookeeperIdGenerator.class);

    private static final String COLON_TOKEN = ":";

    private static final String LINE_TOKEN = "-";

    private static final String PREFIX_PATH = "/snowflake/";

    private static final String SUFFIX_PATH = "/id";

    private String gen;

    private String name;

    private int workerId;

    private CuratorFramework curator;

    private String path;

    private String ipAndPort;

    public ZookeeperIdGenerator(String connectionString, String port,String name) {
        this.ipAndPort = IpUtils.getIp() + COLON_TOKEN + port;
        this.curator = CuratorFrameworkFactory.builder().connectString(connectionString)
                .retryPolicy(new RetryUntilElapsed(1000, 4))
                .connectionTimeoutMs(10000)
                .sessionTimeoutMs(6000)
                .build();
        this.path = PREFIX_PATH+name+SUFFIX_PATH;
    }

    public int getWorkerId() {
        try {
            log.debug("initial zookeeper start");
            this.curator.start();
            Stat stat = this.curator.checkExists().forPath(this.path);
            if (null == stat) {
                String node = createNode();
                log.info("init=> node:{},workerId:{},ipAndPort:{}", node,workerId,ipAndPort);
            } else {
                List<String> nodeList = curator.getChildren().forPath(path);
                String node = nodeList.stream().filter(e -> ipAndPort.equals(parseIpAndPort(e))).findFirst().orElse(null);
                if (null != node) {
                    this.workerId = parseWorkerId(node);
                    log.info("find node=>node:{},workerId:{},ipAndPort:{}", node,workerId,ipAndPort);
                } else {
                    String newNode = createNode();
                    this.workerId = parseWorkerId(newNode);
                    log.info("find node but no workerId=>node:{},workerId:{},ipAndPort:{}", newNode,workerId,ipAndPort);
                }
            }
        } catch (Exception e) {
            log.error("initial zookeeper start ERROR {}", e);
        }
        log.info("this zookeeper can find workerId:{},ipAndPort:{}", workerId,ipAndPort);
        return this.workerId;
    }

    public String parseIpAndPort(String node) {
        return node.split(LINE_TOKEN)[0];
    }

    public Integer parseWorkerId(String node) {
        return Integer.parseInt(node.split(LINE_TOKEN)[1]);
    }

    /**
     * 创建节点
     *
     * @return
     * @throws Exception
     */
    public String createNode() throws Exception {
        String data = this.ipAndPort + System.currentTimeMillis();
        return this.curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path + "/" + ipAndPort + "-", data.getBytes());
    }



    @Override
    public long nextId() {
        int workerId = getWorkerId();
        return new SnowflakeIdGenerator(workerId).nextId();
    }

}
